package org.example.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.example.mapper.SyncConfigMapper;
import org.example.model.SyncConfig;
import org.springframework.jdbc.core.BatchPreparedStatementSetter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class SyncService {

    // 常量定义
    private static final String ALL_TIME_WINDOW = "ALL";
    private static final String TIME_WINDOW_DELIMITER = "-";
    private static final int MAX_CONSECUTIVE_ERRORS = 3;
    private static final String DELETE_FLAG_TRUE = "1";
    private static final String DELETE_FLAG_TRUE_IGNORE_CASE = "true";

    // 数据库操作模板
    private final SyncConfigMapper configMapper;    // 同步配置管理
    private final DataSourceManager dataSourceManager;

    /**
     * 执行单表同步（双库操作）
     * <p>
     * 该方法会执行以下步骤：
     * 1. 检查是否在配置的时间窗口内
     * 2. 从源库查询增量数据
     * 3. 将数据分类为需要删除和需要插入/更新的记录
     * 4. 批量执行删除和插入/更新操作
     * 5. 更新同步状态
     *
     * @param config 同步配置信息
     */
    public void syncTable(SyncConfig config) {
        // 记录开始时间用于性能统计
        LocalDateTime startTime = LocalDateTime.now();

        try {
            // 检查是否在允许执行的时间窗口内
            if (!isInExecutionWindow(config)) {
                log.info("表 {} 跳过同步：不在执行时间窗口内", config.getSourceTable());
                return;
            }

            log.info("开始同步表: {} -> {}", config.getSourceTable(), config.getTargetTable());

            // 计算查询的起始时间：如果有上次成功时间则使用，否则按同步间隔计算
            LocalDateTime lastSuccessTime = config.getLastSuccessTime() != null ?
                    config.getLastSuccessTime() : startTime.minusMinutes(config.getSyncInterval());

            // 从源库查询增量数据
            List<Map<String, Object>> rows = fetchIncrementalDataFromSource(config, lastSuccessTime, startTime);

            // 分类处理数据：分离需要删除和需要插入/更新的记录
            List<Object> deleteKeys = new ArrayList<>();
            List<Map<String, Object>> upsertData = new ArrayList<>();

            for (Map<String, Object> row : rows) {
                if (isDeleted(config, row)) {
                    Object uniqueValue = row.get(config.getUniqueKey());
                    if (uniqueValue != null) {
                        deleteKeys.add(uniqueValue);
                    }
                } else {
                    upsertData.add(row);
                }
            }

            // 批量处理数据
            int processedCount = 0;
            if (!deleteKeys.isEmpty()) {
                processedCount += batchDeleteInTarget(config, deleteKeys);
            }
            if (!upsertData.isEmpty()) {
                processedCount += batchUpsertInTarget(config, upsertData);
            }

            // 更新同步状态到源库（管理库）
            config.setLastSuccessTime(startTime);
            config.setErrorCount(0);
            configMapper.updateSyncStatus(config.getId(), startTime, 0);
            // 使用Duration计算耗时（毫秒）
            LocalDateTime endTime = LocalDateTime.now();
            Duration duration = Duration.between(startTime, endTime);
            log.info("表 {} 同步完成，处理 {} 条记录，耗时 {} ms",
                    config.getSourceTable(),
                    processedCount,
                    duration.toMillis());

        } catch (Exception e) {
            log.error("表 {} 同步失败", config.getSourceTable(), e);
            // 错误计数并更新状态
            int errorCount = config.getErrorCount() + 1;
            config.setErrorCount(errorCount);
            configMapper.updateSyncStatus(config.getId(), config.getLastSuccessTime(), errorCount);

            // 连续错误超过阈值时触发告警
            if (errorCount >= MAX_CONSECUTIVE_ERRORS) {
                log.warn("表 {} 连续同步失败 {} 次，请检查数据源",
                        config.getSourceTable(), errorCount);
            }
        } finally {
            // 无论成功失败都更新最后同步时间
            config.setLastSyncTime(LocalDateTime.now());
            configMapper.updateLastSyncTime(config.getId(), config.getLastSyncTime());
        }
    }

    /**
     * 从源库查询增量数据
     *
     * @param config 同步配置
     * @param start  查询起始时间
     * @param end    查询结束时间
     * @return 增量数据列表
     */
    private List<Map<String, Object>> fetchIncrementalDataFromSource(SyncConfig config,
                                                                     LocalDateTime start,
                                                                     LocalDateTime end) {
        JdbcTemplate sourceJdbcTemplate = dataSourceManager.getJdbcTemplate(config.getSourceDataSourceId());
        int pageSize = 1000; // 每页1000条
        int offset = 0;
        List<Map<String, Object>> allRows = new ArrayList<>();

        while (true) {
            String sql = String.format(
                    "SELECT * FROM %s WHERE %s > ? AND %s <= ? LIMIT ? OFFSET ?",
                    config.getSourceTable(),
                    config.getSyncColumn(),
                    config.getSyncColumn()
            );

            List<Map<String, Object>> page = sourceJdbcTemplate.queryForList(
                    sql, start, end, pageSize, offset
            );

            if (page.isEmpty()) {
                break;
            }

            allRows.addAll(page);
            offset += pageSize;
        }

        return allRows;
    }

    /**
     * 批量执行插入/更新操作
     *
     * @param config     同步配置
     * @param upsertData 需要插入/更新的数据
     * @return 处理的记录数
     */
    private int batchUpsertInTarget(SyncConfig config, List<Map<String, Object>> upsertData) {
        if (upsertData.isEmpty()) {
            return 0;
        }
        NamedParameterJdbcTemplate targetNamedJdbcTemplate =
                dataSourceManager.getNamedJdbcTemplate(config.getTargetDataSourceId());

        // 获取列名（假设所有记录具有相同的列）
        Set<String> columns = upsertData.get(0).keySet();

        // 构建批量UPSERT SQL语句
        String sql = buildBatchUpsertSql(config, columns);

        // 准备批处理参数
        SqlParameterSource[] batchArgs = upsertData.stream()
                .map(MapSqlParameterSource::new)
                .toArray(SqlParameterSource[]::new);

        // 执行批处理操作
        int[] results = targetNamedJdbcTemplate.batchUpdate(sql, batchArgs);

        // 计算实际处理行数
        int processedCount = 0;
        for (int result : results) {
            if (result >= 0) {
                processedCount += result;
            }
        }

        log.debug("批量插入/更新 {} 条记录，表: {}", processedCount, config.getTargetTable());
        return processedCount;
    }

    /**
     * 构建批量UPSERT SQL语句
     *
     * @param config  同步配置
     * @param columns 列名集合
     * @return 完整的UPSERT SQL语句
     */
    private String buildBatchUpsertSql(SyncConfig config, Set<String> columns) {
        String tableName = config.getTargetTable();
        String uniqueKey = config.getUniqueKey();

        // 构建列名和占位符
        String columnsStr = String.join(",", columns);
        String placeholders = columns.stream()
                .map(c -> ":" + c)
                .collect(Collectors.joining(","));

        // 构建更新部分（排除唯一键）
        String updateSet = columns.stream()
                .filter(c -> !c.equals(uniqueKey))
                .map(c -> c + " = VALUES(" + c + ")")
                .collect(Collectors.joining(","));

        // MySQL批量UPSERT语法
        return String.format(
                "INSERT INTO %s (%s) VALUES (%s) " +
                        "ON DUPLICATE KEY UPDATE %s",
                tableName, columnsStr, placeholders, updateSet);
    }

    /**
     * 批量删除目标库记录
     *
     * @param config     同步配置
     * @param deleteKeys 需要删除的记录主键列表
     * @return 删除的记录数
     */
    private int batchDeleteInTarget(SyncConfig config, List<Object> deleteKeys) {
        if (deleteKeys.isEmpty()) {
            return 0;
        }

        // 构建删除SQL
        String deleteSql = String.format("DELETE FROM %s WHERE %s = ?",
                config.getTargetTable(), config.getUniqueKey());

        JdbcTemplate targetJdbcTemplate = dataSourceManager.getJdbcTemplate(config.getTargetDataSourceId());

        // 使用批处理执行删除
        int[] results = targetJdbcTemplate.batchUpdate(
                deleteSql,
                new BatchPreparedStatementSetter() {
                    @Override
                    public void setValues(PreparedStatement ps, int i) throws SQLException {
                        Object key = deleteKeys.get(i);
                        // 根据主键类型设置参数
                        if (key instanceof Long) {
                            ps.setLong(1, (Long) key);
                        } else if (key instanceof String) {
                            ps.setString(1, (String) key);
                        } else if (key instanceof Integer) {
                            ps.setInt(1, (Integer) key);
                        } else {
                            throw new SQLException("不支持的主键类型: " + key.getClass());
                        }
                    }

                    @Override
                    public int getBatchSize() {
                        return deleteKeys.size();
                    }
                });

        // 计算实际删除行数
        int deletedCount = 0;
        for (int result : results) {
            if (result >= 0) {
                deletedCount += result;
            }
        }

        log.debug("批量删除 {} 条记录，表: {}", deletedCount, config.getTargetTable());
        return deletedCount;
    }

    /**
     * 判断记录是否被删除
     *
     * @param config 同步配置
     * @param row    数据行
     * @return true表示记录已被标记为删除，false表示未删除
     */
    private boolean isDeleted(SyncConfig config, Map<String, Object> row) {
        // 如果未配置删除标记列，则默认不删除
        if (config.getDeletedColumn() == null) {
            return false;
        }

        Object deleteFlag = row.get(config.getDeletedColumn());
        if (deleteFlag == null) {
            return false;
        }

        // 处理不同类型的删除标记
        if (deleteFlag instanceof Boolean) {
            return (Boolean) deleteFlag;
        } else if (deleteFlag instanceof Number) {
            return ((Number) deleteFlag).intValue() > 0;
        } else if (deleteFlag instanceof String) {
            return DELETE_FLAG_TRUE.equals(deleteFlag) ||
                    DELETE_FLAG_TRUE_IGNORE_CASE.equalsIgnoreCase((String) deleteFlag);
        }
        return false;
    }

    /**
     * 检查当前时间是否在执行窗口内
     *
     * @param config 同步配置
     * @return true表示可以执行，false表示不在执行窗口内
     */
    boolean isInExecutionWindow(SyncConfig config) {
        // 未配置或配置为ALL表示全天可执行
        if (config.getExecutionWindow() == null ||
                ALL_TIME_WINDOW.equalsIgnoreCase(config.getExecutionWindow())) {
            return true;
        }

        try {
            String window = config.getExecutionWindow().trim();
            String[] parts = window.split(TIME_WINDOW_DELIMITER);
            if (parts.length != 2) {
                // 格式错误默认执行
                return true;
            }

            LocalTime now = LocalTime.now();
            LocalTime start = LocalTime.parse(parts[0].trim());
            LocalTime end = LocalTime.parse(parts[1].trim());

            if (start.isBefore(end)) {
                // 不跨天的时间窗口检查
                return !now.isBefore(start) && !now.isAfter(end);
            } else {
                // 处理跨天情况 (如 22:00-02:00)
                return !now.isBefore(start) || !now.isAfter(end);
            }
        } catch (Exception e) {
            log.error("解析执行窗口失败，默认执行: {}", config.getExecutionWindow(), e);
            return true;
        }
    }
}
